package beautiful.butterfly.drds.data_exchange.taskgroup.runner;


import beautiful.butterfly.drds.data_exchange.plugin.AbstractTaskPlugin;
import beautiful.butterfly.drds.data_exchange.spi.AbstractWriter;
import beautiful.butterfly.drds.data_exchange.statistics.PerfRecord;
import beautiful.butterfly.drds.data_exchange.statistics.Phase;
import beautiful.butterfly.drds.data_exchange.statistics.communication.CommunicationTool;
import beautiful.butterfly.drds.data_exchange.transport.exchanger.Consume;
import org.apache.commons.lang3.Validate;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * 单个slice的writer执行调用
 */
public class WriterRunner extends AbstractRunner implements Runnable
{

    private static final Logger LOG = LoggerFactory
            .getLogger(WriterRunner.class);

    private Consume consume;

    public WriterRunner(AbstractTaskPlugin abstractTaskPlugin)
    {
        super(abstractTaskPlugin);
    }

    public void setConsume(Consume consume)
    {
        this.consume = consume;
    }

    //@Override
    public void run()
    {
        Validate.isTrue(this.consume != null);

        AbstractWriter.AbstractTask abstractTaskWriter = (AbstractWriter.AbstractTask) this.getPlugin();
        //统计waitReadTime，并且在finally end
        PerfRecord channelWaitRead = new PerfRecord(getTaskGroupId(), getTaskId(), Phase.wait_read_time);
        try
        {
            channelWaitRead.start();
            LOG.debug("task writer starts to do init ...");
            PerfRecord initPerfRecord = new PerfRecord(getTaskGroupId(), getTaskId(), Phase.write_task_init);
            initPerfRecord.start();
            abstractTaskWriter.init();
            initPerfRecord.end();

            LOG.debug("task writer starts to do prepare ...");
            PerfRecord preparePerfRecord = new PerfRecord(getTaskGroupId(), getTaskId(), Phase.write_task_prepare);
            preparePerfRecord.start();
            abstractTaskWriter.prepare();
            preparePerfRecord.end();
            LOG.debug("task writer starts to write ...");

            PerfRecord dataPerfRecord = new PerfRecord(getTaskGroupId(), getTaskId(), Phase.write_task_data);
            dataPerfRecord.start();
            abstractTaskWriter.startWrite(consume);

            dataPerfRecord.addCount(CommunicationTool.getTotalReadRecords(super.getRunnerCommunication()));
            dataPerfRecord.addSize(CommunicationTool.getTotalReadBytes(super.getRunnerCommunication()));
            dataPerfRecord.end();

            LOG.debug("task writer starts to do post ...");
            PerfRecord postPerfRecord = new PerfRecord(getTaskGroupId(), getTaskId(), Phase.write_task_post);
            postPerfRecord.start();
            abstractTaskWriter.post();
            postPerfRecord.end();

            super.markSuccess();
        } catch (Throwable e)
        {
            LOG.error("AbstractWriter Runner Received Exceptions:", e);
            super.markFail(e);
        } finally
        {
            LOG.debug("task writer starts to do destroy ...");
            PerfRecord desPerfRecord = new PerfRecord(getTaskGroupId(), getTaskId(), Phase.write_task_destroy);
            desPerfRecord.start();
            super.destroy();
            desPerfRecord.end();
            channelWaitRead.end(super.getRunnerCommunication().getLongCounter(CommunicationTool.wait_reader_time));
        }
    }

    public boolean supportFailOver()
    {
        AbstractWriter.AbstractTask abstractTaskWriter = (AbstractWriter.AbstractTask) this.getPlugin();
        return abstractTaskWriter.supportFailOver();
    }

    public void shutdown()
    {
        consume.shutdown();
    }
}
